package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.BeanToJsonStrFunction;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.bean.DwsExaminationBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.DorisUtil;
import com.atguigu.edu.realtime.util.EnvAndCheckUtils;
import com.atguigu.edu.realtime.util.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

//考试域题目粒度答题各窗口汇总表
public class DwsExaminationQuestionAnswerWindowBean {
    public static void main(String[] args) {
        // TODO 1. 环境准备及状态后端设置
        StreamExecutionEnvironment env = EnvAndCheckUtils.getEnvAndMarkCheck();

        // TODO 2. 从 Kafka dwd_examination_test_question 主题读取数据
        String topic = "dwd_test_exam_question";
        String groupId = "dwd_test_exam_question_answer_window";
        KafkaSource<String> kafkaConsumer = KafkaUtils.getKafkaSource(topic, groupId);
        DataStreamSource<String> source = env.fromSource(kafkaConsumer,
                WatermarkStrategy.noWatermarks(), "dws_question_answer_test_question_source");

        //source.print();
        //{"create_time":"2023-09-11 11:30:34","paper_id":1082,"question_id":2464,
        // "is_correct":"1","score":5.00,"deleted":"0",
        // "answer":"8537,8538","user_id":13,"id":49855,"exam_id":3326,"ts":1694403034}
        // TOPO 3 . 数据格式类型转换

        SingleOutputStreamOperator<DwsExaminationBean> processDs = source.process(new ProcessFunction<String, DwsExaminationBean>() {
            @Override
            public void processElement(String s, ProcessFunction<String, DwsExaminationBean>.Context context, Collector<DwsExaminationBean> collector) throws Exception {
                JSONObject jsonObj = JSON.parseObject(s);
                String isCorrect = jsonObj.getString("is_correct");
                Long correctAnswerCount = isCorrect.equals("1") ? 1L : 0L;
                collector.collect(DwsExaminationBean.builder()
                        .question_id(jsonObj.getString("question_id"))
                        .answer_count(1L)
                        .correctAnswerCount(correctAnswerCount)
                        .ts(jsonObj.getLong("ts") * 1000)
                        .build());
            }
        });
        //processDs.print("格式转换：");
        //DwsExaminationBean(stt=null, edt=null, question_id=2602, question_txt=null, correctAnswerCount=0, answer_count=1, ts=1694403358000)

        //设置水位线
        SingleOutputStreamOperator<DwsExaminationBean> watermarkDs = processDs.assignTimestampsAndWatermarks(WatermarkStrategy.<DwsExaminationBean>forMonotonousTimestamps()
                .withTimestampAssigner(
                        new SerializableTimestampAssigner<DwsExaminationBean>() {
                            @Override
                            public long extractTimestamp(DwsExaminationBean s, long l) {
                                return s.getTs();
                            }
                        }
                ));
        //分组
        KeyedStream<DwsExaminationBean, String> keyByDs = watermarkDs.keyBy(new KeySelector<DwsExaminationBean, String>() {
            @Override
            public String getKey(DwsExaminationBean dwsExaminationBean) throws Exception {
                return dwsExaminationBean.getQuestion_id();
            }
        });
        //开窗
        WindowedStream<DwsExaminationBean, String, TimeWindow> windowDS = keyByDs.window(TumblingEventTimeWindows.of(Time.seconds(3L)));

        //聚合
        SingleOutputStreamOperator<DwsExaminationBean> reduceDs = windowDS.reduce(new ReduceFunction<DwsExaminationBean>() {
                                                                                    @Override
                                                                                    public DwsExaminationBean reduce(DwsExaminationBean value1, DwsExaminationBean value2) throws Exception {
                                                                                        value1.setAnswer_count(value1.getAnswer_count() + value2.getAnswer_count());
                                                                                        value1.setCorrectAnswerCount(value1.getCorrectAnswerCount() + value2.getCorrectAnswerCount());
                                                                                        return value1;

                                                                                    }
                                                                                },
                new ProcessWindowFunction<DwsExaminationBean, DwsExaminationBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<DwsExaminationBean, DwsExaminationBean, String, TimeWindow>.Context context, Iterable<DwsExaminationBean> iterable, Collector<DwsExaminationBean> collector) throws Exception {
                        long start = context.window().getStart();
                        long end = context.window().getEnd();
                        //补充数据
                        for (DwsExaminationBean bean : iterable) {
                            bean.setStt(DateFormatUtil.toYmdHms(start));
                            bean.setCur_date(DateFormatUtil.toDate(start));
                            bean.setEdt(DateFormatUtil.toYmdHms(end));
                            bean.setTs(System.currentTimeMillis());
                            collector.collect(bean);
                        }
                    }
                }
        );

        //关联维度
        SingleOutputStreamOperator<DwsExaminationBean> questionTxtDS = AsyncDataStream.unorderedWait(
                reduceDs,
                new DimAsyncFunction<DwsExaminationBean>("dim_test_question_info") {
                    @Override
                    public void join(DwsExaminationBean obj, JSONObject dimInfoJsonObj) {
                        String questionTxt = dimInfoJsonObj.getString("question_txt");
                        obj.setQuestion_txt(questionTxt);
                    }

                    @Override
                    public String getKey(DwsExaminationBean obj) {
                        return obj.getQuestion_id();
                    }
                },
                60, TimeUnit.SECONDS
        );
        questionTxtDS.print("关联");
        //DwsExaminationBean(stt=2023-09-11 11:49:36, edt=2023-09-11 11:49:39,
        // question_id=741, question_txt=<p>常见的 MQ产品 描述错误的是</p>, correctAnswerCount=2, answer_count=2, ts=1694404201571)

        questionTxtDS.map(
                new BeanToJsonStrFunction<>()
        ).sinkTo(DorisUtil.getDorisSink("dws_examination_question_answer_window"));

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
